package com.yinhai.modules.cluster.domian.jgroups;

import com.yinhai.core.common.api.config.impl.ConfigManager;
import com.yinhai.core.common.api.config.impl.SysConfig;
import com.yinhai.modules.cluster.api.IMessageChannel;
import org.apache.commons.configuration.event.ConfigurationEvent;
import org.apache.commons.configuration.event.ConfigurationListener;
import org.apache.commons.lang3.StringUtils;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.Receiver;
import org.jgroups.protocols.TCPPING;
import org.jgroups.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.net.UnknownHostException;

public  class JMessageChannel implements IMessageChannel<Serializable>,ConfigurationListener{
	private final static Logger logger = LoggerFactory.getLogger(JMessageChannel.class);
	protected JChannel jChannel;
	protected String clusterName;
	protected Receiver receiver;
	private String clusterHosts;
	
	public JMessageChannel(String clusterName, Receiver receiver){
		this.clusterName = clusterName;
		this.receiver = receiver;
	}
	
	@Override
	public void connect() {
		if(this.jChannel == null && SysConfig.getSysConfigToBoolean("isCluster", false) && "TaMessageClusterDWR".equals(this.clusterName)){
			try {
				System.setProperty("java.net.preferIPv4Stack", "true");
				if(SysConfig.getSysConfig("clusterTransport","udp").equals("udp") || !SysConfig.getSysConfigToBoolean("isConfigDistribute",false)){
					this.jChannel = new JChannel("network-udp.xml");
				}else{
					ConfigManager.addConfigurationListener("zkShareConfig",this);
					String localClusterHost = ConfigManager.requestLocalClusterHost(this.clusterName);
					String currentClusterHosts = ConfigManager.getClusterHosts(this.clusterName);
					String bindPort = localClusterHost.substring(localClusterHost.indexOf("[")+1,localClusterHost.indexOf("]"));
					String initialHosts = StringUtils.isBlank(currentClusterHosts)? localClusterHost: (currentClusterHosts.indexOf(localClusterHost) != -1 ? currentClusterHosts : localClusterHost+","+currentClusterHosts);
					System.setProperty("jgroups.tcp.bind_port",bindPort);
					System.setProperty("jgroups.tcpping.initial_hosts", initialHosts);
					this.jChannel = new JChannel("network-tcp.xml");
					//this.jChannel初始化过程中地址可能已经发生变化
					if(!initialHosts.equals(this.clusterHosts)){
						setTCPInitialHosts(this.clusterHosts);
					}
				}
				this.jChannel.setDiscardOwnMessages(true);
				this.jChannel.setReceiver(this.receiver);
				this.jChannel.connect(this.clusterName);
			} catch (Exception e) {
				logger.error("JGroups connect error!",e);
			}
		}
	}

	@Override
	public void configurationChanged(ConfigurationEvent configurationEvent) {
		final String prefix = ConfigManager.CLUSTERHOST +":"+this.clusterName+":";
		String key = configurationEvent.getPropertyName();
		if(key.startsWith(prefix)){
			String currentClusterHosts = ConfigManager.getClusterHosts(this.clusterName);
			setTCPInitialHosts(currentClusterHosts);
			this.clusterHosts = currentClusterHosts;
		}
	}

	private void setTCPInitialHosts(String clusterHosts){
		if(this.jChannel != null){
			TCPPING  ping  = (TCPPING)this.jChannel.getProtocolStack().findProtocol(TCPPING.class);
			if(ping != null){
				try {
					ping.setInitialHosts(Util.parseCommaDelimitedHosts(clusterHosts,0));
				} catch (UnknownHostException e) {
					logger.error("parse clusterHosts fail", e);
				}
			}
		}
	}

	@Override
	public void close(){
		if(this.jChannel != null && !this.jChannel.isClosed()){
			this.jChannel.close();
		}
	}

	@Override
	public void send(Serializable obj){
		if(this.jChannel != null){
			Message msg = new Message();
			msg.setObject(obj);
			msg.setSrc(this.jChannel.getAddress());
			try {
				this.jChannel.send(msg);
			} catch (Exception e) {
				logger.error("JGroups send error!",e);
			}
		}
	}
	
	@Override
	public void send(byte[] bytes){
		if(this.jChannel != null){
			Message msg = new Message();
			msg.setBuffer(bytes);
			msg.setSrc(this.jChannel.getAddress());
			try {
				this.jChannel.send(msg);
			} catch (Exception e) {
				logger.error("JGroups send error!",e);
			}
		}
	}

	public String getClusterName() {
		return clusterName;
	}

	public void setClusterName(String clusterName) {
		this.clusterName = clusterName;
	}
	
	public Receiver getReceiver() {
		return receiver;
	}

	public void setReceiver(Receiver receiver) {
		this.receiver = receiver;
	}

}
